package gae.testjig.task;

import gae.testjig.client.ClientChannelId;
import gae.testjig.data.PMF;
import gae.testjig.dto.Counter;
import gae.testjig.dto.StateMachine;
import gae.testjig.ofy.dao.ODaoSeats;
import gae.testjig.server.ChannelLogger;
import gae.testjig.server.ConstraintTest;
import gae.testjig.server.ConstraintTestStepper;
import gae.testjig.server.DuplicateException;

import java.io.IOException;
import java.util.logging.Logger;

import javax.jdo.PersistenceManager;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import com.google.appengine.api.channel.ChannelMessage;
import com.google.appengine.api.channel.ChannelService;
import com.google.appengine.api.channel.ChannelServiceFactory;
import com.google.appengine.api.datastore.DatastoreService;
import com.google.appengine.api.datastore.DatastoreServiceFactory;
import com.google.appengine.api.datastore.EntityNotFoundException;
import com.google.appengine.api.datastore.Key;
import com.google.appengine.api.datastore.KeyFactory;
import com.google.appengine.api.datastore.Entity;
import com.google.appengine.api.datastore.Transaction;


public class TaskWorkerServlet extends HttpServlet {
	private static final long serialVersionUID = 1L;
	
	final static private int SC_INTERNAL_SERVER_ERROR = 500;
	final static private int MAX_STATE = 30;

	final static private Logger logger = Logger.getLogger(TaskWorkerServlet.class.getSimpleName());
	
	@Override
	protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
		try {
			logger.info("TaskWorkerServlet GET " + req.getRemoteAddr() + ":" + req.getRequestURI());
			String pathInfo = req.getPathInfo();
			if ( pathInfo.startsWith(Task.PATH_ADVANCE_STATE) ) {
				String retryCountString = req.getHeader("X-AppEngine-TaskRetryCount");
				boolean hasRetries = !retryCountString.equals("0");
				
				String clientId = req.getParameter("clientId");
				String channelId = req.getParameter("channelId");
				String stateMachineKeyString = req.getParameter("stateMachineKey");
				String counterKeyString = req.getParameter("counterKey");
				int expectedState = Integer.parseInt(req.getParameter("nextState")); // i'm ignoring NFE exception
				int isDelay = Integer.parseInt(req.getParameter("isDelay")); // i'm ignoring NFE exception
				
				logger.info("advanceState "+expectedState+" "+hasRetries);
				doAdvanceState(ClientChannelId.create(clientId, channelId), stateMachineKeyString, counterKeyString, expectedState, isDelay, hasRetries);
			}
			
			else if ( pathInfo.startsWith(Task.PATH_ADVANCE_STATE_DS) ) {
				String retryCountString = req.getHeader("X-AppEngine-TaskRetryCount");
				boolean hasRetries = !retryCountString.equals("0");
				
				String clientId = req.getParameter("clientId");
				String channelId = req.getParameter("channelId");
				String stateMachineKeyString = req.getParameter("stateMachineKey");
				String counterKeyString = req.getParameter("counterKey");
				int expectedState = Integer.parseInt(req.getParameter("nextState")); // i'm ignoring NFE exception
				int isDelay = Integer.parseInt(req.getParameter("isDelay")); // i'm ignoring NFE exception
				
				logger.info("advanceStateDS "+expectedState+" "+hasRetries);
				doAdvanceStateDS(ClientChannelId.create(clientId, channelId), stateMachineKeyString, counterKeyString, expectedState, isDelay, hasRetries);
			}
	
			else if ( pathInfo.startsWith(Task.PATH_COUNT_INVOCATION) ) {
				String clientId = req.getParameter("clientId");
				String channelId = req.getParameter("channelId");
				String counterKeyString = req.getParameter("counterKey");
				int state = Integer.parseInt(req.getParameter("state")); // i'm ignoring NFE exception
				int invocationCount = Integer.parseInt(req.getParameter("invocationCount")); // i'm ignoring NFE exception
				int retryCount = Integer.parseInt(req.getParameter("retryCount")); // i'm ignoring NFE exception
				int mismatches = Integer.parseInt(req.getParameter("mismatches")); // i'm ignoring NFE exception
	
				logger.info("invocationCount="+invocationCount+" retryCount="+retryCount+" state="+state+" mismatches="+mismatches);
				doCount(ClientChannelId.create(clientId, channelId), counterKeyString, state, invocationCount, retryCount, mismatches);
			}
	
			else if ( pathInfo.startsWith(Task.PATH_COUNT_INVOCATION_DS) ) {
				String clientId = req.getParameter("clientId");
				String channelId = req.getParameter("channelId");
				String counterKeyString = req.getParameter("counterKey");
				int state = Integer.parseInt(req.getParameter("state")); // i'm ignoring NFE exception
				int invocationCount = Integer.parseInt(req.getParameter("invocationCount")); // i'm ignoring NFE exception
				int retryCount = Integer.parseInt(req.getParameter("retryCount")); // i'm ignoring NFE exception
				int mismatches = Integer.parseInt(req.getParameter("mismatches")); // i'm ignoring NFE exception
	
				logger.info("invocationCount="+invocationCount+" retryCount="+retryCount+" state="+state+" mismatches="+mismatches);
				doCountDS(ClientChannelId.create(clientId, channelId), counterKeyString, state, invocationCount, retryCount, mismatches);
			}

			else if ( pathInfo.startsWith(Task.PATH_CONSTRAINT_TEST_RACE_OFY) ) {
				String clientId = req.getParameter("clientId");
				String channelId = req.getParameter("channelId");
				String ownerName = req.getParameter("ownerName");
				String seatId = req.getParameter("seatId");
				ClientChannelId ccid = ClientChannelId.create(clientId, channelId);
				ChannelLogger logger = new ChannelLogger(ccid);
				try {
					ODaoSeats.reserveSeat(logger, ownerName, seatId);
					String info = ODaoSeats.fetchSeatInfo(seatId);
					ChannelService channelService = ChannelServiceFactory.getChannelService();
					channelService.sendMessage(new ChannelMessage(ccid.getClientId(), "{\"channelId\": \""+ccid.getChannelId()+"\", \"info\": \""+info+"\"}"));
				}
				catch (DuplicateException e) {
					logger.info(ownerName+": "+e.getMessage());
					logger.info(ownerName+": ABORT");
				}
				catch (Exception e) {
					logger.info(ownerName+": exception: "+e.getMessage());
					logger.info(ownerName+": ABORT");
				}
			}

			else if ( pathInfo.startsWith(Task.PATH_CONSTRAINT_TEST_RACE) ) {
				String clientId = req.getParameter("clientId");
				String channelId = req.getParameter("channelId");
				String ownerName = req.getParameter("ownerName");
				String seatId = req.getParameter("seatId");
				ClientChannelId ccid = ClientChannelId.create(clientId, channelId);
				ChannelLogger logger = new ChannelLogger(ccid);
				ConstraintTest ct = new ConstraintTest();
				try {
					ct.reserveSeat(logger, ownerName, seatId);
					String info = ct.fetchSeatInfo(seatId);
					ChannelService channelService = ChannelServiceFactory.getChannelService();
					channelService.sendMessage(new ChannelMessage(ccid.getClientId(), "{\"channelId\": \""+ccid.getChannelId()+"\", \"info\": \""+info+"\"}"));
				}
				catch (DuplicateException e) {
					logger.info(ownerName+": "+e.getMessage());
					logger.info(ownerName+": ABORT");
				}
				catch (Exception e) {
					logger.info(ownerName+": exception: "+e.getMessage());
					logger.info(ownerName+": ABORT");
				}
			}
			
			else if ( pathInfo.startsWith(Task.PATH_CONSTRAINT_TEST) ) {
				String clientId = req.getParameter("clientId");
				String channelId = req.getParameter("channelId");
				String ownerName = req.getParameter("ownerName");
				String seatId = req.getParameter("seatId");
				new ConstraintTestStepper(ClientChannelId.create(clientId, channelId), seatId, ownerName).performSteps();
			}
			
		}
		catch (java.util.ConcurrentModificationException e) {
			resp.setStatus(SC_INTERNAL_SERVER_ERROR); // let the queue retry again later.
		}
	}

	private void doAdvanceState(ClientChannelId ccid, String stateMachineKeyString, String counterKeyString, int expectedState, int isDelay, boolean hasRetriesInHeader) {
		Key stateMachineKey = KeyFactory.stringToKey(stateMachineKeyString);
		PersistenceManager pm = PMF.getForTest1().getPersistenceManager();
		Task taskBuilder = Task.BUILD;
		if ( isDelay != 0 ) {
			pm = PMF.getForTest2().getPersistenceManager();
			taskBuilder = Task.DELAY2;
		}
		pm.currentTransaction().begin();
		try {
			StateMachine stateMachine = pm.getObjectById(StateMachine.class, stateMachineKey);
			int state = stateMachine.getState();
			if ( state != expectedState ) {
				logger.info("state="+state+" expected="+expectedState);
				Task.COUNTER_QUEUE.add(taskBuilder.countInvocation(ccid, counterKeyString, 0, 0, 0, 1));
			}
			if ( state < MAX_STATE ) {
				// advance the state machine by one.
				int nextState = 1 + state;
				stateMachine.setState(nextState);
				pm.makePersistent(stateMachine);
				
				// queue the task to advance to the next state.
				Task.TX_TEST_QUEUE.add(taskBuilder.advanceState(ccid, stateMachineKeyString, counterKeyString, nextState, isDelay));
				
				// record our visit.
				Task.COUNTER_QUEUE.add(taskBuilder.countInvocation(ccid, counterKeyString, state, 1, (hasRetriesInHeader ? 1 : 0), 0));
			}

			pm.currentTransaction().commit();
		}
		finally {
			if (pm.currentTransaction().isActive()) {
		        pm.currentTransaction().rollback();
		    }
			pm.close();
		}
	}
	
	private void doAdvanceStateDS(ClientChannelId ccid, String stateMachineKeyString, String counterKeyString, int expectedState, int isDelay, boolean hasRetriesInHeader) {
		Key stateMachineKey = KeyFactory.stringToKey(stateMachineKeyString);
		
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();

		Task taskBuilder = Task.BUILD;
//		if ( isDelay != 0 ) {
//			taskBuilder = Task.DELAY2;
//		}

		Transaction txn = datastore.beginTransaction();
		try {
			Entity stateMachineEntity = datastore.get(stateMachineKey);
			Long stateInteger = (Long) stateMachineEntity.getProperty("state");
			int state = stateInteger.intValue();
			if ( state != expectedState ) {
				logger.info("state="+state+" expected="+expectedState);
				Task.COUNTER_QUEUE.add(txn, taskBuilder.countInvocationDS(ccid, counterKeyString, 0, 0, 0, 1));
			}
			if ( state < MAX_STATE ) {
				// advance the state machine by one.
				int nextState = 1 + state;
				stateMachineEntity.setProperty("state", nextState);
				datastore.put(txn, stateMachineEntity);
				
				// queue the task to advance to the next state.
				Task.TX_TEST_QUEUE.add(txn, taskBuilder.advanceStateDS(ccid, stateMachineKeyString, counterKeyString, nextState, isDelay));
				
				// record our visit.
				Task.COUNTER_QUEUE.add(txn, taskBuilder.countInvocationDS(ccid, counterKeyString, state, 1, (hasRetriesInHeader ? 1 : 0), 0));
			}

			txn.commit();
		}
		catch (EntityNotFoundException e) {
			logger.severe("stateMachine entity not found by key "+stateMachineKeyString+". "+e.getMessage());
		}
		finally {
			if (txn.isActive()) {
		        txn.rollback();
		    }
		}
	}
	
	private void doCount(ClientChannelId ccid, String counterKeyString, int state, int invocationCount, int retryCount, int mismatches) {
		StringBuilder buf = new StringBuilder();
		buf.append("{");
		buf.append("\"channelId\": \"").append(ccid.getChannelId()).append("\", ");
		buf.append("\"state\": ").append(state).append(", ");
		buf.append("\"invocationCount\": ").append(invocationCount).append(", ");
		buf.append("\"retryCount\": ").append(retryCount).append(", ");
		buf.append("\"mismatches\": ").append(mismatches);
		buf.append("}");

		logger.info("doCount: clientId="+ccid.getClientId()+" buf="+buf.toString());
		ChannelService channelService = ChannelServiceFactory.getChannelService();
		channelService.sendMessage(new ChannelMessage(ccid.getClientId(), buf.toString()));
		
		Key counterKey = KeyFactory.stringToKey(counterKeyString);
		PersistenceManager pm = PMF.getForTransactions().getPersistenceManager();
		pm.currentTransaction().begin();
		try {
			Counter counter = pm.getObjectById(Counter.class, counterKey);
			counter.setInvocationCount(invocationCount + counter.getInvocationCount());
			counter.setRetryCount(retryCount + counter.getRetryCount());
			pm.makePersistent(counter);
			pm.currentTransaction().commit();
		}
		finally {
			if (pm.currentTransaction().isActive()) {
		        pm.currentTransaction().rollback();
		    }
			pm.close();
		}
	}
	
	private void doCountDS(ClientChannelId ccid, String counterKeyString, int state, int invocationCount, int retryCount, int mismatches) {
		StringBuilder buf = new StringBuilder();
		buf.append("{");
		buf.append("\"channelId\": \"").append(ccid.getChannelId()).append("\", ");
		buf.append("\"state\": ").append(state).append(", ");
		buf.append("\"invocationCount\": ").append(invocationCount).append(", ");
		buf.append("\"retryCount\": ").append(retryCount).append(", ");
		buf.append("\"mismatches\": ").append(mismatches);
		buf.append("}");

		logger.info("doCount: clientId="+ccid.getClientId()+" buf="+buf.toString());
		ChannelService channelService = ChannelServiceFactory.getChannelService();
		channelService.sendMessage(new ChannelMessage(ccid.getClientId(), buf.toString()));
		
		Key counterKey = KeyFactory.stringToKey(counterKeyString);
		
		DatastoreService datastore = DatastoreServiceFactory.getDatastoreService();
		Transaction txn = datastore.beginTransaction();
		try {
			Entity counterEntity = datastore.get(counterKey);
			Long currentCountInteger = (Long) counterEntity.getProperty("invocationCount");
			int currentCount = currentCountInteger.intValue();
			counterEntity.setProperty("invocationCount", new Long(currentCount + invocationCount));
			Long currentRetryInteger = (Long) counterEntity.getProperty("retryCount");
			int currentRetry = currentRetryInteger.intValue();
			counterEntity.setProperty("retryCount", new Long(currentRetry + retryCount));
			datastore.put(counterEntity);
			txn.commit();
		}
		catch (EntityNotFoundException e) {
			logger.severe("counter entity not found by key "+counterKeyString+". "+e.getMessage());
		}
		finally {
			if (txn.isActive()) {
		        txn.rollback();
		    }
		}
	}
	
}
